package com.im.service.websocket.session;

import com.alibaba.fastjson.JSON;
import com.im.service.infra.entity.ChatHistory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class SessionManager {
    private SessionManager() {}
    private static final Map<String, WebSocketSession> SESSION_POOL = new ConcurrentHashMap<>();

    public static void putSession(String userId, WebSocketSession session) {
        SESSION_POOL.put(userId, session);
    }

    public static void removeSession(String userId) {
        log.info("用户：【{}】已下线", userId);
        SESSION_POOL.remove(userId);
    }



    public static Sender getSender(String userId) {
        return new Sender(SESSION_POOL.get(userId));
    }

    public static class Sender {
        private final WebSocketSession webSocketSession;

        public Sender(WebSocketSession webSocketSession) {
            this.webSocketSession = webSocketSession;
        }

        public Mono<ChatHistory> sendMessage(ChatHistory chatHistory) {
            String messageContent = JSON.toJSONString(chatHistory);
            if (isAvailable()) {
                return webSocketSession.send(Mono.just(webSocketSession.textMessage(messageContent)))
                        .then(Mono.just(chatHistory))
                        .doOnNext(it -> it.setForwardSuccessTime(LocalDateTime.now()))
                        .doOnTerminate(() -> log.info("消息转发成功，message ：{}", messageContent));
            } else {
                return Mono.just(chatHistory)
                        .doOnTerminate(() -> log.info("用户不在线，将转为离线消息，message ：{}", messageContent));
            }
        }

        public boolean isAvailable() {
            return webSocketSession != null && webSocketSession.isOpen();
        }
    }
}
